package com.example.kafkaspringboot.controller;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RestController;

import java.util.concurrent.ExecutionException;

@RestController
public class KafkaProducer {
    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    // 发送消息
    @GetMapping("/kafka/normal/{message}")
    public void sendMessage1(@PathVariable("message") String normalMessage) {
        kafkaTemplate.send("topic-test", "0", normalMessage).addCallback(success ->{
            System.out.println("发送消息成功：" + normalMessage );
        }, failure -> {
            System.out.println("发送消息失败");
        });
    }

    // 发送消息
    @GetMapping("/kafka/normal/{message}")
    public void sendMessage2(@PathVariable("message") String normalMessage) {
        ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send("topic-test", "0", normalMessage);
        try {
            SendResult<String, Object> result = future.get();
            System.out.println(result);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
    }

    // 发送消息
    @GetMapping("/kafka/normal/topic1/{message}")
    public void sendMessage3(@PathVariable("message") String normalMessage) {
        kafkaTemplate.send("topic1", normalMessage).
                addCallback(
                        success -> System.out.println("发送消息成功：" + normalMessage ),
                        failure -> System.out.println("发送消息失败"));
    }
}